package org.niit.service

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

class ZhangBatchService {

  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
  val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  spark.sparkContext.setLogLevel("ERROR")

  def dataAnalysis(): Unit = {

    val data = spark.read.option("header", "true").csv("input/data_jd.csv")

//    data.show()

    ZhangBatch1(data)
    ZhangBatch2(data)
    ZhangBatch3(data)
    ZhangBatch4(data)
    ZhangBatch5(data)
    ZhangBatch6(data)

    spark.stop()


  }

  //需求1：分析不同年龄区间的新用户占比情况。
  def ZhangBatch1(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |-- 分析不同年龄区间的新用户占比
        |WITH age_group_data AS (
        |    -- 第一步：划分年龄区间，并筛选新用户（new_user=1）
        |    SELECT
        |        user_id,
        |        age,
        |        -- 按常见年龄分段划分区间（可根据业务调整分段）
        |        CASE
        |            WHEN age < 18 THEN '0-17岁'
        |            WHEN age BETWEEN 18 AND 24 THEN '18-24岁'
        |            WHEN age BETWEEN 25 AND 30 THEN '25-30岁'
        |            WHEN age BETWEEN 31 AND 40 THEN '31-40岁'
        |            WHEN age BETWEEN 41 AND 50 THEN '41-50岁'
        |            ELSE '51岁以上'
        |        END AS age_group
        |    FROM
        |        data  -- 替换为你的表名
        |    WHERE
        |        new_user = 1  -- 只统计新用户
        |),
        |total_new_users AS (
        |    -- 第二步：计算新用户总数量（用于计算占比）
        |    SELECT COUNT(DISTINCT user_id) AS total_new_user_count
        |    FROM age_group_data
        |)
        |-- 第三步：统计各年龄区间新用户数量及占比
        |SELECT
        |    age_group,
        |    COUNT(DISTINCT user_id) AS new_user_count,  -- 各区间新用户数
        |    -- 计算占比（保留2位小数）
        |    ROUND(
        |        (COUNT(DISTINCT user_id) * 1.0 / (SELECT total_new_user_count FROM total_new_users)) * 100,
        |        2
        |    ) AS new_user_ratio_percent  -- 占比（百分比）
        |FROM
        |    age_group_data
        |GROUP BY
        |    age_group
        |ORDER BY
        |    -- 按年龄区间升序排序（需按分段逻辑手动指定顺序）
        |    CASE age_group
        |        WHEN '0-17岁' THEN 1
        |        WHEN '18-24岁' THEN 2
        |        WHEN '25-30岁' THEN 3
        |        WHEN '31-40岁' THEN 4
        |        WHEN '41-50岁' THEN 5
        |        WHEN '51岁以上' THEN 6
        |    END;
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch1.csv") // 输出目录，内部包含一个 CSV 文件


  }

  //需求2：研究不同性别用户在各市场级别下的分布
  def ZhangBatch2(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |-- 分析不同性别用户在各市场级别下的分布
        |WITH gender_market_data AS (
        |    -- 统计每个性别在各市场级别的用户数（去重）
        |    SELECT
        |        sex,
        |        market,
        |        COUNT(DISTINCT user_id) AS user_count  -- 每个性别+市场级别的用户数量
        |    FROM
        |        data  -- 替换为你的表名
        |    GROUP BY
        |        sex, market
        |),
        |total_users_by_gender AS (
        |    -- 计算每个性别的总用户数（用于计算性别内的市场分布占比）
        |    SELECT
        |        sex,
        |        SUM(user_count) AS total_gender_users
        |    FROM
        |        gender_market_data
        |    GROUP BY
        |        sex
        |)
        |-- 最终结果：性别、市场级别、用户数、该性别下的市场占比
        |SELECT
        |    gmd.sex,
        |    gmd.market,
        |    gmd.user_count,
        |    -- 计算该市场级别用户在对应性别中的占比（保留2位小数，百分比）
        |    ROUND(
        |        (gmd.user_count * 1.0 / tug.total_gender_users) * 100,
        |        2
        |    ) AS ratio_in_gender_percent
        |FROM
        |    gender_market_data gmd
        |LEFT JOIN
        |    total_users_by_gender tug
        |    ON gmd.sex = tug.sex
        |ORDER BY
        |    gmd.sex, gmd.market;  -- 按性别、市场级别排序
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch2.csv") // 输出目录，内部包含一个 CSV 文件


  }

  //需求3：比较不同设备上用户的平均浏览总页面数
  def ZhangBatch3(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |-- 比较不同设备上用户的平均浏览总页面数
        |SELECT
        |    device,  -- 设备类型（如mobile、desktop等）
        |    COUNT(DISTINCT user_id) AS user_count,  -- 各设备的用户数量（去重）
        |    ROUND(AVG(total_pages_visited), 2) AS avg_total_pages  -- 平均浏览总页面数（保留2位小数）
        |FROM
        |    data  -- 替换为你的表名
        |GROUP BY
        |    device  -- 按设备分组
        |ORDER BY
        |    avg_total_pages DESC;  -- 按平均页面数降序排列（可改为ASC升序）
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch3.csv") // 输出目录，内部包含一个 CSV 文件


  }

  //需求4：分析新用户和老用户在不同设备上的浏览总页面数差异
  def ZhangBatch4(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |-- 分析新用户和老用户在不同设备上的浏览总页面数差异
        |SELECT
        |  -- 将new_user转换为用户类型标签（1=新用户，0=老用户）
        |  CASE
        |    WHEN new_user = '1' THEN '新用户'
        |    WHEN new_user = '0' THEN '老用户'
        |    ELSE '未知用户'
        |  END AS user_type,
        |  device AS device_type,  -- 设备类型（直接使用device字段）
        |  -- 计算该分组的平均浏览总页数（单用户浏览深度）
        |  ROUND(AVG(total_pages_visited), 2) AS avg_total_pages,
        |  -- 计算该分组的总浏览页数（群体整体浏览规模）
        |  SUM(total_pages_visited) AS sum_total_pages,
        |  -- 统计该分组的去重用户数（样本量参考）
        |  COUNT(DISTINCT user_id) AS user_count,
        |  -- 计算该设备在该用户类型中的浏览页数占比（辅助分析设备权重）
        |  ROUND(
        |    SUM(total_pages_visited) / SUM(SUM(total_pages_visited)) OVER (PARTITION BY new_user),
        |    4
        |  ) AS device_page_ratio
        |FROM
        |  data  -- 替换为你的表名（原始数据注册的Spark表名）
        |GROUP BY
        |  new_user, device  -- 按用户类型（new_user）和设备（device）分组
        |ORDER BY
        |  user_type, device_type;  -- 按用户类型和设备排序
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch4.csv") // 输出目录，内部包含一个 CSV 文件


  }

  //需求5：研究不同市场级别下不同操作系统用户的平均浏览页数和支付率
  def ZhangBatch5(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |--研究不同市场级别下不同操作系统用户的平均浏览页数和支付率
        |SELECT
        |  market AS market_level,  -- 市场级别（对应market字段）
        |  operative_system AS os_type,  -- 操作系统类型（对应operative_system字段）
        |  COUNT(DISTINCT user_id) AS unique_users,  -- 去重用户数
        |  ROUND(AVG(total_pages_visited), 2) AS avg_total_pages,  -- 平均浏览总页数
        |  -- 计算支付页到支付确认页的转化率（保留2位小数）
        |  ROUND(
        |    -- 分子：访问支付页且完成支付确认的去重用户数
        |    COUNT(DISTINCT CASE WHEN payment_page = '1' AND payment_confirmation_page = '1' THEN user_id END)
        |    -- 分母：访问支付页的去重用户数（避免除以0）
        |    / NULLIF(COUNT(DISTINCT CASE WHEN payment_page = '1' THEN user_id END), 0) * 100,
        |    2
        |  ) AS payment_to_confirmation_rate
        |FROM
        |  data  -- 替换为你的表名（原始数据注册的Spark表）
        |GROUP BY
        |  market, operative_system  -- 按市场级别和操作系统分组
        |ORDER BY
        |  market_level, os_type;  -- 按市场级别和操作系统排序
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch5.csv") // 输出目录，内部包含一个 CSV 文件


  }

  //需求6：研究新用户在不同市场级别下的来源渠道偏好。
  def ZhangBatch6(data: DataFrame):Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ZhangBatchService")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    data.createOrReplaceTempView("data")

    val sql =
      """
        |-- 研究新用户在不同市场级别下的来源渠道偏好
        |SELECT
        |    market,
        |    source,
        |    COUNT(DISTINCT user_id) AS new_user_count,
        |    ROUND(
        |        (COUNT(DISTINCT user_id) * 1.0 /
        |         (SELECT COUNT(DISTINCT user_id)
        |          FROM data
        |          WHERE new_user = 1 AND market = data.market)) * 100,
        |        2
        |    ) AS channel_ratio_in_market
        |FROM
        |    data AS outer_query  -- 增加表别名，避免子查询中market字段歧义
        |WHERE
        |    new_user = 1
        |GROUP BY
        |    market, source
        |ORDER BY
        |    market, channel_ratio_in_market DESC;
        |""".stripMargin

    val frame = spark.sql(sql)

    frame.show()

    frame.coalesce(1) // 将所有数据合并到一个分区
      .write
      .format("csv")
      .option("header", "true")
      .option("delimiter", ",")
      .option("encoding", "UTF-8")
      .mode("overwrite")
      .save("output/ZhangBatch6.csv") // 输出目录，内部包含一个 CSV 文件


  }


}
